package com.audaque.springboot.foshanupload.web.kafkademo.controller;

import cn.hutool.json.JSONUtil;
import com.audaque.springboot.foshanupload.web.kafkademo.model.rsp.SysUserMpgtRsp;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * @author zgb
 * @desc ...
 * @date 2023-04-13 16:31:46
 */
@Slf4j
@RestController
@RequestMapping("/kafka/producer")
public class ProducerController {

    @Resource
    private KafkaTemplate<Object, Object> kafkaTemplate;


    /**
     * 发送到特定主题
     */
    @PostMapping("/simpleSend/{topic}")
    public String simpleSend(@PathVariable("topic") String topic,
                             @RequestBody SysUserMpgtRsp sysUserMpgtRsp) {
        //指定好kafka的topic之后，调用send方法会自动帮我们创建好topic，只是创建的topic默认是1个副本和1个分区的，数量可以在kafka的server.properties修改，也可以在客户端注册主题bean
        String msg = JSONUtil.toJsonStr(sysUserMpgtRsp);
        kafkaTemplate.send(topic, msg);
        return "Kafka 生产消息成功~";
    }

    /**
     * 发送到特定主题、特定分区:在发送一条消息时，可以指定这条消息的key，Producer根据这个key和Partition机制来判断应该将这条消息发送到哪个Parition。
     */
    @GetMapping("/simpleSend1/{topic}/{key}/{msg}")
    public String simpleSend1(@PathVariable("topic") String topic, @PathVariable("key") String key,
                              @PathVariable("msg") String msg) {
        kafkaTemplate.send(topic, key, msg);
        return "Kafka 生产消息成功~";
    }


    /**
     * 发送到特定主题
     *
     * @param topic 主题
     * @param msg   消息
     * @return 结果
     */
    @GetMapping("/simpleSend/{topic}/{msg}")
    public String simpleSend(@PathVariable("topic") String topic,
                             @PathVariable("msg") String msg) {
        //指定好kafka的topic之后，调用send方法会自动帮我们创建好topic，只是创建的topic默认是1个副本和1个分区的，数量可以在kafka的server.properties修改，也可以在客户端注册主题bean
        kafkaTemplate.send(topic, msg);
        return "Kafka 生产消息成功~";
    }


    /**
     * 同步生成消息
     *
     * @param topic 主题
     * @param msg   消息
     * @return 结果
     */
    @GetMapping("/synSend/{topic}/{msg}")
    public String synSend(@PathVariable("topic") String topic,
                          @PathVariable("msg") String msg) throws Exception {
        //指定好kafka的topic之后，调用send方法会自动帮我们创建好topic，只是创建的topic默认是1个副本和1个分区的，数量可以在kafka的server.properties修改，也可以在客户端注册主题bean
        kafkaTemplate.send(topic, msg).get(3, TimeUnit.MINUTES);
        return "Kafka 生产消息成功~";
    }


    /**
     * 带回调的发送消息
     *
     * @param topic 主题
     * @param msg   消息
     * @return 结果
     */
    @GetMapping("/callbackSend/{topic}/{msg}")
    public String callbackSend(@PathVariable("topic") String topic,
                               @PathVariable("msg") String msg) {
        try {
            // 发送消息并设置成功和失败的逻辑
            kafkaTemplate.send(topic, msg).addCallback(new ListenableFutureCallback<SendResult<Object, Object>>() {
                // 发送失败后处理逻辑
                @Override
                public void onFailure(Throwable e) {
                    System.out.println("消息发送失败..." + e.getMessage());
                }

                // 发送成功后处理逻辑
                @Override
                public void onSuccess(SendResult<Object, Object> result) {
                    System.out.println("消息发送成功..." + result.getRecordMetadata().topic() + "-" +
                            result.getRecordMetadata().partition() + "-" + result.getRecordMetadata().offset());
                }
            });
        } catch (Throwable throwable) {
            return "kafka 发送消息失败~";
        }
        return "kafka 发送消息成功~";
    }


    /**
     * kafka 事务发送消息
     *
     * @param topic 主题
     * @param msg   消息
     * @return 结果
     */
    @GetMapping("/transactionSend/{topic}/{msg}")
    public String transactionSend(@PathVariable("topic") String topic,
                                  @PathVariable("msg") String msg) {
        // 声明事务：后面报错消息不会发出去
        kafkaTemplate.executeInTransaction(operations -> {
            operations.send(topic, msg);
            //异常要抛出方法才能触发消息回滚
            throw new RuntimeException("fail");
            //return true;
        });

        return "成功";
    }

    @GetMapping("/batch")
    public void batch() {
        for (int i = 0; i < 12; i++) {
            kafkaTemplate.send("topic.quick.batch", "test batch listener,dataNum-" + i);
        }
    }
}
